import math, re, sys, os, MySQLdb, shutil, hashlib
from functools import wraps
from datetime import datetime
from collections import defaultdict
from contextlib import closing
import cardsharp as cs
from cardsharp.errors import LoadError
from configuration import config, db_info
from errors import CalcError
__all__ = ['RUNDATETIME', 'FORMAT_INFO', 'FORMAT_DICT', 'PHASE_MAP',
'key_as_str', 'check_file', 'calc', 're_escape', 'get_state_key',
'get_cnxn', 'get_conv_cnxn', 'is_file', 'is_none', 'memoize',
'str_to_boolean', 'get_state_name', 'get_max_review_round', 'copy_to_review',
'set_LOOKUPS', 'LOOKUPS', 'check_format','capture_input', 'get_hash']
RUNDATETIME = datetime.now().isoformat('_').replace(':', '-')
FORMAT_INFO = {
#'sas' : '.sas7bdat',
'text' : '.txt',
'excel' : '.xls',
'spss' : '.sav',
'csharp' : '.csharp',
'csv' : '.csv',
#'sas7bdat' : '.sas7bdat',
'txt' : '.txt',
'xls' : '.xls',
'sav' : '.sav',
'del' : '.del',
#'.sas7bdat' : 'sas',
'.txt' : 'text',
'.xls' : 'excel',
'.sav' : 'spss',
'.csharp' : 'csharp',
'.csv' : 'csv',
#'.sas7bdat' : 'sas7bdat',
'.del' : 'del',
}
FORMAT_DICT = {'float':float,
'integer':int,
'boolean':bool,
'string':str,
}
PHASE_MAP = {0: 'stage',
1: 'pre_process',
2: 'process',
3: 'transform',
4: 'compare'}
try:
region_ds = cs.load(source=os.path.join(db_info['stand']['dir'], r'lookups\region.xls'), format='excel')
region_ds.variables['id'].convert('integer')
except LoadError:
region_ds = None
print 'Unable to load region dataset'
regions = {}
LOOKUPS = defaultdict(list)
[docs]def get_hash(key):
return hashlib.md5(key).hexdigest()
[docs]def set_LOOKUPS(db_info):
for file in os.listdir(os.path.join(config.meta_dir, 'database', 'standardized', 'lookups')):
lookup_name = file.replace('.xls', '')
with closing(MySQLdb.connect(host=config.db_host, user=db_info['user'], passwd=db_info['pass'],
charset='utf8', db=db_info['name'], port=config.db_port, use_unicode=True)) as conn:
with closing(conn.cursor()) as cur:
cur.execute('SELECT id from %s' % lookup_name)
for row in cur.fetchall():
LOOKUPS[lookup_name].append(row[0])
if region_ds:
for row in region_ds:
regions[row['id']] = row['abbr']
for row in region_ds:
regions[row['abbr']] = row['id']
[docs]def get_cnxn(db_name, opt):
return MySQLdb.connect(host=config.db_host, user=opt['db_info']['user'], passwd=opt['db_info']['pass'],
charset='utf8', db=db_name, port=config.db_port, use_unicode=True)
[docs]def get_conv_cnxn(db_name, opt):
from MySQLdb.converters import conversions
return MySQLdb.connect(host=config.db_host, user=opt['db_info']['user'], passwd=opt['db_info']['pass'],
charset='utf8', db=db_name, port=config.db_port, use_unicode=True, conv=conversions)
[docs]def key_as_str(value, inner_trim=True):
"""Produces a key value for looking up in a crosswalk.
If value is None than returns an empty string.
>>> v = None
>>> key_as_str(v) + 'XXX'
XXX
If value is a basestring removes leading and trailing whitespace if inner_trim = True,
multiple line spaces made into one space, value is lowered:
>>> v = ' DS sS '
>>> key_as_str(v)
ds ss
If inner_trime is false just strim leading and trailing and lower
>>> v = ' DF sdsD'
>>> v
df sdsd
"""
if value is None:
return ''
else:
if isinstance(value, basestring):
if inner_trim:
return ' '.join(value.split()).lower()
else:
return value.strip().lower()
else:
return str(value)
[docs]def get_state_key(value):
if not value:
return ''
else:
if isinstance(value, basestring):
try:
#check to see if the string passed in is an int
return str(regions.get(int(value)))
except:
return value.lower().strip()
else:
return str(regions.get(value))
[docs]def get_state_name(value):
if not value:
return ''
else:
if isinstance(value, basestring):
try:
#check to see if the string passed in is an int
return str(regions.get(int(value)))
except:
return str(regions.get(value))
return str(regions.get(value))
[docs]def check_file(file_path, as_dir = False):
"""Helper function to check if file or directory is valid."""
if not os.path.exists(file_path):
print "%s does not exist." % file_path
sys.exit(1)
if as_dir:
if not os.path.isdir(file_path):
print "%s is not a directory." % file_path
sys.exit(1)
else:
if not os.path.isfile(file_path):
print "%s is not a file." % file_path
sys.exit(1)
return file_path
#get the highest round for a state
[docs]def get_max_review_round(segment, region, set=config.current_set, **kw):
_path = os.path.join(config.review_dir, 'set_%s' % set, segment, region)
#check to see if no review files exist
#if no files exist return lowest to allow creation of initial review round
if not os.path.isdir(_path):
return -1
round = -1
_round = -1
for dir in os.listdir(_path):
try:
_round = int(re.search('round(\d+)', dir).groups()[0])
except AttributeError, ValueError:
print 'warning: bad review folder %s' % os.path.join(_path, dir)
if _round > round:
round = _round
max_dir = dir
if kw.get('as_dir'):
return os.path.join(_path, max_dir)
else:
return round
[docs]def copy_to_review(name, region, orig_path):
"""Copy output file to review directory and return the path to the review directory"""
_round = get_max_review_round(name, region)
_round = 1 if _round == -1 else _round + 1
_review_path = os.path.join(config.review_dir, 'set_%s' % config.current_set, name, region, 'round%s' % _round)
if not os.path.isdir(_review_path): os.makedirs(_review_path)
shutil.copy(orig_path, _review_path)
return _review_path
[docs]def is_file(file_path):
if not os.path.isfile(file_path):
return False
return True
integers_regex = re.compile(r'\b[\d\.]+\b')
def paranreplace(matchobj):
return re.sub('\(|\)(?!\()', lambda a: '*(' if a.group(0) == '(' else ')*', matchobj.group(0))
parans = re.compile('\d\(|\)\(|\)\d')
[docs]def calc(expr, advanced=False):
def safe_eval(expr, symbols={}):
expr = re.sub(parans, paranreplace, expr)
return eval(expr, dict(__builtins__=None), symbols)
def whole_number_to_float(match):
group = match.group()
if group.find('.') == -1:
return group + '.0'
return group
expr = expr.replace('^','**')
expr = expr.replace(',','')
expr = integers_regex.sub(whole_number_to_float, expr)
try:
if advanced:
return safe_eval(expr, vars(math))
else:
return safe_eval(expr)
except:
raise CalcError
_escape_map = {'\\' : '\\\\',
'.' : '\\.',
'^' : '\\^',
'$' : '\\$',
'*' : '\\*',
'+' : '\\+',
'?' : '\\?',
'{' : '\\{',
'}' : '\\}',
'[' : '\\[',
']' : '\\]',
'|' : '\\|',
'(' : '\\(',
')' : '\\)'}
[docs]def re_escape(value):
'''
Escape a regular expression.
>>> re_escape(ab(#))
u'ab\((\d+|#\d+|#)\)'
'''
return ''.join([_escape_map.get(c, c) for c in value])
[docs]def is_none(value):
return True if value is None or value is '' else False
[docs]def memoize(func):
"""This function integrates memoization, an optimization technique, into cardsharp.
Memoize wraps a function with a single argument. When it is called the first time, the result
is cached with the argument as the key. If it is called in the future with the same
argument, it returns the cached value.
>>> from random import random
>>> func = lambda x: random()
>>> func(1) == func(1)
False
>>> func = memoize(func)
>>> func(1) == func(1)
True
"""
memo = dict()
@wraps(func)
def _func(arg):
if arg in memo:
return memo[arg]
memo[arg] = value = func(arg)
return value
return _func
[docs]def str_to_boolean(s):
"""Converts a string to a boolean.
Convert map:
-------------------
| 1 | *True* |
-------------------
| true | *True* |
-------------------
| 0 | *False* |
-------------------
| false | *False* |
-------------------
If *s* is *None* returns *None* instead.
>>> str_to_boolean('1')
True
>>> str_to_boolean(0)
False
>>> str_to_boolean(' FAlse ')
False
>>> str_to_boolean('3')
Traceback (most recent call last):
...
FormatError: Invalid input data, 3: must be true, false, 1 or 0
"""
if s is None:
return None
else:
if str(s) in ['1', '1.0'] or str(s).strip().lower() == 'true':
return True
elif str(s) in ['0', '0.0'] or str(s).strip().lower() == 'false':
return False
else:
raise FormatError('Invalid input data, %s: must be true, false, 1 or 0' % s)